#include <stx/imp/cmn.hpp>

inline namespace stx
{
class Actor__ final
{
    Task fn_task_;

public:
    Actor__() = default;

    Actor__(Task fn_task) : fn_task_(std::move(fn_task))
    {}

    Actor__& operator=(Task fn_task)
    {
        this->fn_task_ = std::move(fn_task);
        return *this;
    }

    void run()
    {
        if (this->fn_task_)
        {
            this->fn_task_();
        }
    }
};

using SpActor__ = std::shared_ptr<Actor__>;

struct EpollerTask__ final
{
    SpActor__ sp_actor;
    Fd        fd       = -1;
    FdState   fd_state = {};
    Duration  timeout  = 0ms;
};

GLOBAL_VAR(gTasks__, BlockingLockFreeQueue<Task>{});
GLOBAL_VAR(gEpollerTasks__, BlockingLockFreeQueue<EpollerTask__>{});
GLOBAL_VAR(gEpollController__, EventFd{});

Future<FdState>
wait(Fd const fd, FdState const fd_state, Duration const dur_timeout)
{
    auto const tp_start = SteadyClock::now();

    {
        thread_local auto tls_epoller =
            FileDescriptor(::epoll_create1(EPOLL_CLOEXEC));
        if (tls_epoller.fd() < 0)
        {
            LOG_ERROR("Failed to epoll_create1: "sv,
                      tls_epoller.fd(),
                      _I_,
                      fd,
                      _I_,
                      fd_state,
                      _I_,
                      M_SYS_ERR);
            std::exit(EXIT_FAILURE);
        }

        auto ee = ::epoll_event{};
        if (isReadable(fd_state))
        {
            ee.events |= EPOLLIN;
        }

        if (isWritable(fd_state))
        {
            ee.events |= EPOLLOUT;
        }

        if (::epoll_ctl(tls_epoller.fd(), EPOLL_CTL_ADD, fd, &ee) < 0)
        {
            LOG_ERROR("EPOLL_CTL_ADD: "sv,
                      tls_epoller.fd(),
                      _I_,
                      fd,
                      _I_,
                      fd_state,
                      _I_,
                      M_SYS_ERR);
            co_return FdState::K_SYS_ERROR;
        }

        auto status = ::epoll_wait(tls_epoller.fd(), &ee, 1, 0);
        ON_SCOPE_EXIT({
            if (::epoll_ctl(tls_epoller.fd(), EPOLL_CTL_DEL, fd, nullptr) < 0)
            {
                LOG_ERROR("EPOLL_CTL_DEL: "sv,
                          tls_epoller.fd(),
                          _I_,
                          fd,
                          _I_,
                          fd_state,
                          _I_,
                          M_SYS_ERR);
            }
        });

        if (status > 0)
        {
            ASSERT(1 == status);
            co_return fd_state;
        }

        if (status < 0)
        {
            LOG_ERROR("Failed to epoll_wait: "sv,
                      tls_epoller.fd(),
                      _I_,
                      fd,
                      _I_,
                      fd_state,
                      _I_,
                      M_SYS_ERR);
            co_return FdState::K_SYS_ERROR;
        }
    }

    auto const sp_prom = std::make_shared<Promise<>>();
    gEpollerTasks__().enqueue(EpollerTask__{
        .sp_actor = std::make_shared<Actor__>([=] { sp_prom->setValue(); }),
        .fd       = fd,
        .fd_state = fd_state,
        .timeout  = dur_timeout});
    gEpollController__().notify();

    co_await sp_prom->getFuture();
    if (dur_timeout > 0ms && elapsed(tp_start) >= dur_timeout)
    {
        co_return FdState::K_TIMEOUT;
    }

    co_return fd_state;
}

Future<FdState> wait(Fd const fd, FdState const fd_state)
{
    return wait(fd, fd_state, 0ms);
}

void schedule(Task fn_task)
{
    gTasks__().enqueue(std::move(fn_task));
}

class Epoller__ final : public FileDescriptor
{
    using SpTimer_ = std::shared_ptr<TimerFd>;

    struct EntryInfo_ final
    {
        SpActor__ sp_actor  = {};
        SpTimer_  sp_timer  = {};
        UniqueId  uid_buddy = {};
        Fd        fd_main   = -1;
        Fd        fd_timer  = -1;
    };

    UniqueId                           uid_epoll_ctler_ = getUniqueId();
    UnorderedMap<UniqueId, EntryInfo_> ee_map_;
    PodBuffer<::epoll_event>           ees_;

public:
    Epoller__() : ees_(128)
    {
        this->fd_ = ::epoll_create1(EPOLL_CLOEXEC);
        if (this->fd_ < 0)
        {
            RUNTIME_ERROR("Failed to ::epoll_create1(): "sv, M_SYS_ERR);
        }

        auto ee     = ::epoll_event{};
        ee.events   = EPOLLIN | EPOLLWAKEUP;
        ee.data.u64 = this->uid_epoll_ctler_;
        this->addEntry_(gEpollController__().fd(), ee);
    }

    NON_COPYABLE(Epoller__);
    NON_MOVABLE(Epoller__);

    [[noreturn]] void loop()
    {
        auto const fn_wait_for_events = [this]
        {
            auto const n = ::epoll_wait(
                this->fd_, this->ees_.data(), Number(this->ees_.size()), -1);
            if (n < 0)
            {
                if (EINTR == errno)
                {
                    return;
                }

                LOG_ERROR("::epoll_wait: "sv, this->fd_, _I_, M_SYS_ERR);
                return;
            }

            if (0 == n)
            {
                LOG_ERROR("::epoll_wait: timeout: "sv, this->fd_);
                return;
            }

            gEpollController__().clear();
            this->handleEvents_(Number(n));
            this->fetchNewPendingActors_();
        };

        while (true)
        {
            fn_wait_for_events();
        }
    }

private:
    bool addEntry_(Fd const fd, ::epoll_event& ee)
    {
        if (::epoll_ctl(this->fd_, EPOLL_CTL_ADD, fd, &ee) < 0)
        {
            LOG_ERROR("EPOLL_CTL_ADD: "sv, fd, _I_, M_SYS_ERR);
            return false;
        }

        return true;
    }

    void removeEntry_(Fd const fd)
    {
        if (::epoll_ctl(this->fd_, EPOLL_CTL_DEL, fd, nullptr) < 0)
        {
            LOG_ERROR("EPOLL_CTL_DEL: "sv, fd, _I_, M_SYS_ERR);
        }
    }

    void handleEvents_(std::size_t const evt_cnt)
    {
        auto const p_ees = this->ees_.data();

        LOOP (i, evt_cnt)
        {
            auto const& ee  = p_ees[i];
            auto const  uid = UniqueId(ee.data.u64);

            if (uid == this->uid_epoll_ctler_)
            {
                continue;
            }

            auto const pos = this->ee_map_.find(uid);
            if (pos == this->ee_map_.end())
            {
                continue;
            }

            auto const entry_info = pos->second;

            this->ee_map_.erase(entry_info.uid_buddy);
            this->ee_map_.erase(uid);
            this->removeEntry_(entry_info.fd_main);
            this->removeEntry_(entry_info.fd_timer);

            schedule([sp_actor = entry_info.sp_actor] { sp_actor->run(); });
        }
    }

    void fetchNewPendingActors_()
    {
        auto epoller_task = EpollerTask__{};
        while (gEpollerTasks__().try_dequeue(epoller_task))
        {
            this->addEpollerTask_(&epoller_task);
        }
    }

    void addEpollerTask_(EpollerTask__* p_epoller_task)
    {
        auto sp_timer = std::make_shared<TimerFd>();

        if (p_epoller_task->timeout > 0ms)
        {
            sp_timer->reset(p_epoller_task->timeout, p_epoller_task->timeout);
        }
        else
        {
            sp_timer->reset(gMaxTimeout(), gMaxTimeout());
        }

        auto const fd_main   = p_epoller_task->fd;
        auto const fd_timer  = sp_timer->fd();
        auto const uid_main  = getUniqueId();
        auto const uid_timer = getUniqueId();

        this->ee_map_[uid_main] =
            EntryInfo_{.sp_actor  = p_epoller_task->sp_actor,
                       .sp_timer  = sp_timer,
                       .uid_buddy = uid_timer,
                       .fd_main   = fd_main,
                       .fd_timer  = fd_timer};

        this->ee_map_[uid_timer] =
            EntryInfo_{.sp_actor  = p_epoller_task->sp_actor,
                       .sp_timer  = sp_timer,
                       .uid_buddy = uid_main,
                       .fd_main   = fd_main,
                       .fd_timer  = fd_timer};

        auto const cmn_flags =
            EPOLLET | EPOLLONESHOT | EPOLLRDHUP | EPOLLWAKEUP;

        auto ee   = ::epoll_event{};
        ee.events = EPOLLIN | cmn_flags;

        ee.data.u64 = uid_timer;
        this->addEntry_(fd_timer, ee);

        switch (p_epoller_task->fd_state)
        {
        case FdState::K_READABLE_OR_WRITABLE:
            ee.events = EPOLLIN | EPOLLOUT | cmn_flags;
            break;
        case FdState::K_READABLE:
            ee.events = EPOLLIN | cmn_flags;
            break;
        case FdState::K_WRITABLE:
            ee.events = EPOLLOUT | cmn_flags;
            break;
        default:
            ee.events = cmn_flags;
        }

        ee.data.u64 = uid_main;
        this->addEntry_(fd_main, ee);
    }
};

void thrdWorker__()
{
    while (true)
    {
        try
        {
            auto fn_task = Task{};
            gTasks__().wait_dequeue(fn_task);
            fn_task();
        }
        catch (std::exception const& e)
        {
            LOG_ERROR("Task error: "sv, e.what());
        }
        catch (...)
        {
            LOG_ERROR("Unknown task error."sv);
        }
    }
}

void startEventLoop(std::size_t const worker_thrd_cnt,
                    std::string const s_thrd_tag)
{
    LOOP (i, worker_thrd_cnt)
    {
        createThread(scat("worker_"sv, i), thrdWorker__);
    }

    setOsThreadTag(s_thrd_tag.data());
    Epoller__{}.loop();
}

void startEventLoop(std::size_t const worker_thrd_cnt)
{
    startEventLoop(worker_thrd_cnt, {});
}

void startEventLoop(std::string s_thrd_tag)
{
    startEventLoop(std::max(2uz, getTotalNumberOfCores()),
                   std::move(s_thrd_tag));
}

void startEventLoop()
{
    startEventLoop(std::max(2uz, getTotalNumberOfCores()), {});
}

TimerController::TimerController(SbRepeating const sb_repeating)
    : sp_timer_state_(std::make_shared<std::atomic<TimerState>>(
          sb_repeating ? TimerState::K_REPEATING : TimerState::K_ONESHOT))
{}

TimerState TimerController::getTimerState() const noexcept
{
    return *this->sp_timer_state_;
}

void TimerController::cancel() noexcept
{
    auto expected = TimerState::K_REPEATING;
    this->sp_timer_state_->compare_exchange_weak(expected,
                                                 TimerState::K_CANCELED);
}

Future<> startTimerTask(Duration const  dur_initial_expiration,
                        Duration const  dur_repeated_expiration,
                        Task            fn_task,
                        TimerController timer_ctler)
{
    ASSERT(fn_task);

    auto tfd = TimerFd(dur_initial_expiration, dur_repeated_expiration);

    while (true)
    {
        auto const fd_state = co_await wait(tfd.fd(), FdState::K_READABLE);
        if (!isReadable(fd_state))
        {
            LOG_ERROR("Invalid fd_state: "sv, fd_state);
            co_return;
        }

        if (TimerState::K_CANCELED == *timer_ctler.sp_timer_state_)
        {
            co_return;
        }

        if (TimerState::K_ONESHOT == *timer_ctler.sp_timer_state_)
        {
            fn_task();
            co_return;
        }

        if (TimerState::K_REPEATING == *timer_ctler.sp_timer_state_)
        {
            fn_task();
            tfd.reset(dur_repeated_expiration, dur_repeated_expiration);
            continue;
        }

        RUNTIME_ERROR();
    }
}

Future<> startTimerTask(Duration const& dur_interval,
                        Task            fn_task,
                        TimerController timer_ctler)
{
    return startTimerTask(
        dur_interval, dur_interval, std::move(fn_task), timer_ctler);
}

Future<> startTimerTask(Duration const dur_interval, Task fn_task)
{
    return startTimerTask(dur_interval,
                          dur_interval,
                          std::move(fn_task),
                          TimerController(SbRepeating(false)));
}

Future<> startRepeatingTimerTask(Duration const dur_interval, Task fn_task)
{
    return startTimerTask(
        dur_interval, dur_interval, std::move(fn_task), TimerController{});
}

Future<> startOneshotTimerTask(Duration const dur_interval, Task fn_task)
{
    return startTimerTask(dur_interval,
                          dur_interval,
                          std::move(fn_task),
                          TimerController(SbRepeating(false)));
}
} // namespace stx

/*
[Q]: Will closing a file descriptor cause it to be removed from all epoll
sets automatically?

[A]: Yes, but be aware of the following point. A file descriptor is a
reference to an open file description (see open(2)). Whenever a descriptor is
duplicated via dup(2), dup2(2), fcntl(2) F_DUPFD, or fork(2), a new file
descriptor referring to the same open file description is created. An open
file description continues to exist until all file descriptors referring to
it have been closed. A file descriptor is removed from an epoll set only
after all the file descriptors referring to the underlying open file
description have been closed (or before if the descriptor is explicitly
removed using epoll_ctl(2) EPOLL_CTL_DEL). This means that even after a file
descriptor that is part of an epoll set has been closed, events may be
reported for that file descriptor if other file descriptors referring to the
same underlying file description remain open.

EPOLLHUP means you can't write, EPOLLRDHUP means you can't read.

 [local end]: shutdown(SHUT_WR):FIN_WAIT_1 [Immediately trigger] EPOLLHUP
 [local end]: close(fd) [then if] [local end]: read/write -> EPOLLERR
[remote end]: close(fd) [then if] [local end]: read/write -> EPOLLERR
[remote end]: shutdown(SHUT_WR):FIN_WAIT_1 [Immediately trigger] EPOLLRDHUP

`man pipe`:
    pipefd[0] refers to the read end of the pipe. pipefd[1] refers to the
    write end of the pipe.

    If all file descriptors referring to the write end of a pipe have been
    closed, then an attempt to read(2) from the pipe will see end-of-file
    (read(2) will return 0). If all file descriptors referring to the read
    end of a pipe have been closed, then a write(2) will cause a SIGPIPE
    signal to be generated for the calling process.  If the calling process
    is ignoring this signal, then write(2) fails with the error EPIPE. An
    application that uses pipe(2) and fork(2) should use suitable close(2)
    calls to close unnecessary duplicate file descriptors; this ensures that
    end-of-file and SIGPIPE/EPIPE are delivered when appropriate.

{
    constexpr auto C_READ_END  = 0;
    constexpr auto C_WRITE_END = 1;

    auto pipefd = std::array<int, 2>{};
    ::pipe(pipefd.data());
    ::close(pipefd[C_READ_END]);

    ::write(pipefd[C_WRITE_END], "hello", 5);
    ASSERT(EPIPE == errno);
}

When fd is added into epoll, if its input buffer has data, then EPOLLIN will
be triggered, no matter whether EPOLLET is specified.

When fd is added into epoll, if the remote end has closed it, then EPOLLRDHUP
will be triggered, no matter whether EPOLLET is specified.
*/